package com.reactor.冷热序列;

import java.time.Duration;
import java.time.temporal.ChronoUnit;

import reactor.core.publisher.Flux;

public class Demo {

	public static void main(String[] args) {
		// 之前的代码清单中所创建的都是冷序列。冷序列的含义是不论订阅者在何时订阅该序列，总是能收到序列中产生的全部消息。此外，对于冷序列来说，没有订阅就没有数据产生
		// 而与之对应的热序列，则是在持续不断地产生消息，订阅者只能获取到在其订阅之后产生的消息。
		// 原始的序列中包含 10 个间隔为 1 秒的元素。通过publish()方法把一个Flux对象转换成ConnectableFlux对象。方法autoConnect()的作用是当ConnectableFlux对象有一个订阅者时就开始产生消息。
		// 代码source.subscribe()的作用是订阅该ConnectableFlux对象，让其开始产生数据。
		// 接着当前线程睡眠 5 秒钟，第二个订阅者此时只能获得到该序列中的后 5 个元素，因此所输出的是数字 5 到 9。
		final Flux<Long> source = Flux.interval(Duration.of(1, ChronoUnit.SECONDS))
		        .take(10)
		        .publish()
		        .autoConnect();
		source.subscribe();
		try {
			Thread.sleep(5000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		source
		        .toStream()
		        .forEach(System.out::println);
	}

}
